package com.shujia.dx

import org.apache.flink.api.common.functions.{ReduceFunction, RuntimeContext}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{MapState, MapStateDescriptor, ReducingState, ReducingStateDescriptor, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

object Demo02CityTouristCntOnFlow {
  def main(args: Array[String]): Unit = {

    // 读取kafka的数据
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    // 指定Kafka Broker集群的地址
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    // 指定消费者组ID
    properties.setProperty("group.id", "dianxin")

    val flinkKafkaConsumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("dianxin", new SimpleStringSchema(), properties)

    flinkKafkaConsumer.setStartFromEarliest() // 从头开始消费
    //    flinkKafkaConsumer.setStartFromLatest()        // 从最新的数据开始消费，如果当前组是第一次消费 也会从头开始消费数据
    //    flinkKafkaConsumer.setStartFromTimestamp(...)  // 从某个时间点开始消费
    //    flinkKafkaConsumer.setStartFromGroupOffsets()  // 默认的 使用组的偏移量进行消费

    // 将Kafka的Consumer注册为Source -- 无界流
    val kafkaDS: DataStream[String] = env.addSource(flinkKafkaConsumer)

    val cityCntDS: DataStream[(String, Long)] = kafkaDS
      .map(line => {
        val arr: Array[String] = line.split(",")
        // 将mdn以及city_id取出
        (arr(0), arr(2))
      })
      .keyBy(_._2)
      .process(new KeyedProcessFunction[String, (String, String), (String, Long)] {
        var mapState: MapState[String, String] = _
        var reducingState: ReducingState[Long] = _

        override def open(parameters: Configuration): Unit = {
          val ctx: RuntimeContext = getRuntimeContext

          // 构建MapState用于对每个城市的游客按照mdn进行去重
          mapState = ctx.getMapState(new MapStateDescriptor[String, String]("mdns", classOf[String], classOf[String]))
          // 构建ReducingState用于统计每个城市的游客数
          reducingState = ctx.getReducingState(new ReducingStateDescriptor[Long]("cnt", new ReduceFunction[Long] {
            override def reduce(value1: Long, value2: Long): Long = {
              value1 + value2
            }
          }, classOf[Long]))

        }

        // 每来一条数据会处理一次
        override def processElement(value: (String, String),
                                    ctx: KeyedProcessFunction[String, (String, String), (String, Long)]#Context,
                                    out: Collector[(String, Long)]): Unit = {
          val (mdn, cityId): (String, String) = value
          if (!mapState.contains(mdn)) {
            mapState.put(mdn, mdn)
            reducingState.add(1L)

            val cnt: Long = reducingState.get()
            out.collect((cityId, cnt))
          }

        }
      })

    cityCntDS
      .addSink(new RichSinkFunction[(String, Long)] {
        var conn: Connection = _
        var preSt: PreparedStatement = _

        override def open(parameters: Configuration): Unit = {
          /**
           * CREATE TABLE `city_cnt` (
           * `city_id` varchar(255) NOT NULL,
           * `cnt` bigint(20) NOT NULL,
           * PRIMARY KEY (`city_id`)
           * ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
           */
          // 建立连接
          Class.forName("com.mysql.jdbc.Driver")
          conn = DriverManager.getConnection("jdbc:mysql://master:3306/crm", "root", "123456")
          preSt = conn.prepareStatement("replace into city_cnt values(?,?)")

        }

        override def close(): Unit = {
          preSt.close()
          conn.close()
        }

        override def invoke(value: (String, Long), context: SinkFunction.Context[_]): Unit = {
          val (cityId, cnt): (String, Long) = value
          preSt.setString(1, cityId)
          preSt.setLong(2, cnt)

          preSt.execute()


        }
      })


    env.execute()
  }


}
